package com.xzx.spark.core.exercise

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 *
 * 需求一：Top10 热门品类
 * 先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
 *
 * @author xinzhixuan
 * @version 1.0
 * @date 2021-08-08 2:14 下午
 */
object Exercise02 {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName(getClass.getSimpleName))
    val rdd = sc.textFile("src/main/scala/com/xzx/spark/core/exercise/user_visit_action.txt")
      .map(x => {
        val data = x.split("_")
        UserVisitAction(data(0), data(1).toLong, data(2), data(3).toLong, data(4), data(5), data(6).toLong, data(7).toLong, data(8), data(9), data(10), data(11), data(12).toLong)
      }).cache()
    //    rdd.foreach(println)
    //实现方案一
    //    method1(rdd)
    //实现方案二，减少shuffle
    //    method2(rdd)
    //方案三：对方案二再进行一次简写
    //    method3(rdd)
    // 方案四：使用累加器，直接规避shuffle
    method4(sc, rdd)

    sc.stop()
  }

  def method4(sc:SparkContext, rdd: RDD[UserVisitAction]): Unit = {
    val acc = new UserVisitActionAcc
    sc.register(acc, "myAcc")
    rdd.foreach(x => {
      acc.add(x)
    })
    acc.value.toList.sortBy(x=>x._2)(Ordering.Tuple3(Ordering.Int.reverse, Ordering.Int.reverse, Ordering.Int.reverse)).take(10).foreach(println)

  }

  def method3(rdd: RDD[UserVisitAction]): Unit = {
    val result = rdd.flatMap(x => {
      if (x.click_category_id != -1) {
        List((x.click_category_id, (1, 0, 0)))
      } else if (x.order_category_ids != "null") {
        val tuples: Array[(Long, (Int, Int, Int))] = x.order_category_ids.split(",").map(x => (x.toLong, (0, 1, 0)))
        tuples.toList
      } else if (x.pay_category_ids != "null") {
        val tuples: Array[(Long, (Int, Int, Int))] = x.pay_category_ids.split(",").map(x => (x.toLong, (0, 0, 1)))
        tuples.toList
      } else {
        Nil
      }
    })
      .reduceByKey {
        case (t1, t2) => (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    result.sortBy(_._2, ascending = false).take(10).foreach(println)
  }

  def method2(rdd: RDD[UserVisitAction]): Unit = {
    // 点击次数
    val clickRDD: RDD[(Long, (Int, Int, Int))] = rdd.filter(_.click_category_id != -1).map(x => (x.click_category_id, (1, 0, 0)))
    // 下单次数
    val orderRDD: RDD[(Long, (Int, Int, Int))] = rdd.filter(_.order_category_ids != "null").flatMap(x => x.order_category_ids.split(",").map(x => (x.toLong, (0, 1, 0))))
    // 支付次数
    val payRDD: RDD[(Long, (Int, Int, Int))] = rdd.filter(_.pay_category_ids != "null").flatMap(x => x.pay_category_ids.split(",").map(x => (x.toLong, (0, 0, 1))))
    val result: RDD[(Long, (Int, Int, Int))] = clickRDD.union(orderRDD).union(payRDD).reduceByKey {
      case (t1, t2) => (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
    }
    result.sortBy(_._2, ascending = false).take(10).foreach(println)
  }

  def method1(rdd: RDD[UserVisitAction]): Unit = {
    //分别统计每个品类点击的次数，下单的次数和支付的次数：
    // 点击次数
    val clickRDD: RDD[(Long, Int)] = rdd.filter(_.click_category_id != -1).map(x => (x.click_category_id, 1)).reduceByKey(_ + _)
    // 下单次数
    val orderRDD: RDD[(Long, Int)] = rdd.filter(_.order_category_ids != "null").flatMap(x => x.order_category_ids.split(",").map(x => (x.toLong, 1))).reduceByKey(_ + _)
    // 支付次数
    val payRDD: RDD[(Long, Int)] = rdd.filter(_.pay_category_ids != "null").flatMap(x => x.pay_category_ids.split(",").map(x => (x.toLong, 1))).reduceByKey(_ + _)
    //三个rdd关联在一起，可以用的算子有  join zip leftOuterJoin cogroup
    // join算子不可行，因为有可能有下单，但是没支付
    // zip算子，要求分区数相同，分区内的元素也相同，这个不可行
    // leftOuterJoin算子，有种场景是就是有偶下单，但是没点击，比如跨天的数据
    // cogroup算子，没有特殊的过多要求，可行
    val result: RDD[(Long, (Int, Int, Int))] = clickRDD.cogroup(orderRDD, payRDD).mapValues {
      case (clickItr, orderItr, payItr) =>
        var clickCnt = 0
        var orderCnt = 0
        var payCnt = 0
        clickItr.foreach(clickCnt += _) // 这里itr其实里面只有一个元素
        orderItr.foreach(orderCnt += _)
        payItr.foreach(payCnt += _)
        (clickCnt, orderCnt, payCnt)
    }
    result.sortBy(_._2, ascending = false).take(10).foreach(println)
  }
}

class UserVisitActionAcc extends AccumulatorV2[UserVisitAction, mutable.Map[Long, (Int, Int, Int)]] {
  private val map = mutable.HashMap[Long, (Int, Int, Int)]()

  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[UserVisitAction, mutable.Map[Long, (Int, Int, Int)]] = new UserVisitActionAcc()

  override def reset(): Unit = map.clear()

  override def add(v: UserVisitAction): Unit = {
    if (v.click_category_id != -1) {
      val tuple = map.getOrElse(v.click_category_id, (0, 0, 0))
      map.update(v.click_category_id, (tuple._1 + 1, tuple._2, tuple._3))
    } else if (v.order_category_ids != "null") {
      v.order_category_ids.split(",").foreach(x => {
        val tuple = map.getOrElse(x.toLong, (0, 0, 0))
        map.update(x.toLong, (tuple._1, tuple._2 + 1, tuple._3))
      })
    } else if (v.pay_category_ids != "null") {
      v.pay_category_ids.split(",").foreach(x => {
        val tuple = map.getOrElse(x.toLong, (0, 0, 0))
        map.update(x.toLong, (tuple._1, tuple._2, tuple._3 + 1))
      })
    }
  }

  override def merge(other: AccumulatorV2[UserVisitAction, mutable.Map[Long, (Int, Int, Int)]]): Unit = {
    other.value.foreach {
      case (category_id, (clickCnt, orderCnt, payCnt)) => {
        val tuple = map.getOrElse(category_id, (0, 0, 0))
        map.update(category_id, (tuple._1 + clickCnt, tuple._2 + orderCnt, tuple._3 + payCnt))
      }
    }
  }

  override def value: mutable.Map[Long, (Int, Int, Int)] = map
}

//用户访问动作表
case class UserVisitAction(
                            date: String, //用户点击行为的日期
                            user_id: Long, //用户的 ID
                            session_id: String, //Session 的 ID
                            page_id: Long, //某个页面的 ID
                            action_time: String, //动作的时间点
                            search_keyword: String, //用户搜索的关键词
                            click_category_id: Long, //某一个商品品类的 ID
                            click_product_id: Long, //某一个商品的 ID
                            order_category_ids: String, //一次订单中所有品类的 ID 集合
                            order_product_ids: String, //一次订单中所有商品的 ID 集合
                            pay_category_ids: String, //一次支付中所有品类的 ID 集合
                            pay_product_ids: String, //一次支付中所有商品的 ID 集合
                            city_id: Long //城市 id
                          )